package com.headfirst.dmp.report.ProvinceCityRpt

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
  * 读取未转成parquet之前的原始数据（sparkContext也无法读取parquet文件）
  *
  * 统计各省市的数据量分布情况
  *
  * （本质上这就是一个分组统计）
  */
object ProvinceCityRPT_CORE {

  def main(args: Array[String]): Unit = {

    //1.判断参数个数
    if (args.length != 2) {
      print(
        """
          |com.hedfirst.dmp.report.ProvinceCityRPT_CORE
          |参数：
          |    logInputPath
          |    resultOutputPath
        """.stripMargin)
      sys.exit(-1)
    }

    //2.接受参数
    val Array(logInputPath, resultOutputPath) = args

    //3.创建SparkContext对象
    val conf = new SparkConf()
    conf.setAppName(s"${this.getClass.getSimpleName}")
    conf.setMaster("local[*]")
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")

    val sc = new SparkContext(conf)


    //4.读取并处理文件(切分-->过滤-->合并-->聚合-->整理格式)
    val result: RDD[String] = sc.textFile(logInputPath)
                                .map(line => line.split(",", -1))
                                .filter(_.length >= 85)
                                .map(arr => {
                                  ((arr(24), arr(25)), 1)
                                })
                                .reduceByKey(_ + _)
                                .map(x => {
                                  x._1._1 + "," + x._1._2 + "," + x._2
                                })

    //5.将数据输出成文件
    result.saveAsTextFile(resultOutputPath)


    // 也可尝试将数据通过JDBC的形式写入mysql （当然也可以通过spark-sql的形式，先组装成DataFrame再写入）
//    val arrResult: Array[String] = result.collect()
//    val url="jdbc:mysql://127.0.0.1:3306/db_scala_ip_province?useUnicode=true&characterEncoding=utf8&serverTimezone=GMT&useSSL=false"
//    val user="root"
//    val password="tiger"
//    val sql="insert into spark_sql3(province,city,count) values(?,?,?)"
//    val connection: Connection = DriverManager.getConnection(url,user,password)
//    val ps: PreparedStatement = connection.prepareStatement(sql)
//    arrResult.foreach(x => {
//      val arr: Array[String] = x.split(",")
//      ps.setString(1,arr(0))
//      ps.setString(2,arr(1))
//      ps.setString(3,arr(2))
//      ps.addBatch()
//    })
//    ps.executeBatch()    //此处使用的是批量插入
//    if(ps!=null){
//      ps.close()
//    }
//    if(connection!=null){
//      connection.close()
//    }


    //6.关流
    sc.stop()

  }

}
